package com.base.test.mysql2es.repository;

import com.base.test.mysql2es.util.CollectionUtil;
import com.base.test.mysql2es.util.ObjectUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

@Slf4j
@Component
@RequiredArgsConstructor
public class EsRepository {

    @Value("${db2es.enable:true}")
    private boolean enable;

    private final RestHighLevelClient client;

    public int saveDataToEs(String index, Map<String, Map<String, String>> dataMap) {
        if (!enable) {
            return 0;
        }

        if (CollectionUtil.isEmpty(dataMap)) {
            return 0;
        }

        BulkRequest batchRequest = new BulkRequest();
        long originalSize = 0;
        for (Map.Entry<String, Map<String, String>> entry : dataMap.entrySet()) {
            String id = entry.getKey();
            Map<String, String> source = entry.getValue();
            if (ObjectUtil.isNotBlank(id) && CollectionUtil.isNotEmpty(source)) {
                IndexRequest doc = new IndexRequest(index).id(id);
                String data = source.get("data");
                if (ObjectUtil.isNotBlank(data)) {
                    doc.source(data, XContentType.JSON);
                    String routing = source.get("routing");
                    if (ObjectUtil.isNotBlank(routing)) {
                        doc.routing(routing);
                    }
                    Long version = ObjectUtil.toLong(source.get("version"));
                    if (ObjectUtil.greater0(version)) {
                        doc.versionType(VersionType.EXTERNAL_GTE).version(version);
                    }
                    batchRequest.add(doc);
                    originalSize++;
                }
            }
        }

        try {
            BulkResponse responses = client.bulk(batchRequest, RequestOptions.DEFAULT);
            int size = responses.getItems().length;

            for (BulkItemResponse response : responses) {
                if (response.isFailed()) {
                    BulkItemResponse.Failure failure = response.getFailure();
                    if (!failure.getMessage().contains("version_conflict_engine_exception")) {
                        if (log.isErrorEnabled()) {
                            log.error("batch save({}) size({}) success({}), has error({})",
                                    index, originalSize, size, failure);
                        }
                    }
                    size--;
                }
            }
            if (size == originalSize) {
                if (log.isDebugEnabled()) {
                    log.debug("batch save({}) size({}) success({})", index, originalSize, size);
                }
            }
            return size;
        } catch (IOException e) {
            if (log.isErrorEnabled()) {
                log.error(String.format("create or update (%s) es data exception", index), e);
            }
            return 0;
        }
    }
}
